package com.openflag.bwicservice.messagequeue;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.openflag.bwicservice.domain.BidRequest;
import com.openflag.bwicservice.domain.CancelRequest;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutionException;

@Component
public class BwicProducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    private static final String BID_TOPIC = "bid-topic";
    private static final String CANCEL_TOPIC = "cancel-topic";
    //
    public void sendBid(BidRequest bidRequest) {
        try {
            kafkaTemplate.send(BID_TOPIC, object2Json(bidRequest)).get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    public void sendCancel(CancelRequest cancelRequest) {
        try {
            kafkaTemplate.send(CANCEL_TOPIC, object2Json(cancelRequest)).get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    private String object2Json(Object obj) {
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            return objectMapper.writeValueAsString(obj);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }
}
